package com.zjl.SpringBoot.第21章_流式_响应式_编程.C_Reactive响应式核心;

import com.zjl.util.OutColour;
import org.junit.Test;
import org.reactivestreams.Subscription;
import reactor.core.Disposable;
import reactor.core.publisher.Flux;
import reactor.core.publisher.BaseSubscriber;
import reactor.core.publisher.SignalType;

import java.io.IOException;

/**
 * ClassName: D2_Reactor实现之_Flux高级
 * Package: com.zjl.SpringBoot.第21章_流式_响应式_编程.C_Reactive响应式核心
 * Description:
 *
 * @Author 张蛟龙
 * @Create 2025/3/19 20:44
 * @Version 1.0
 */
public class D4_Flux高级之订阅者消费者 {
    @Test
    public void Flux的订阅方法() throws IOException {
        Flux<Integer> flux = Flux.range(1, 5)
                .map(i -> {
                    if (i == 3) {
//                        i = i / 0;
                    }
                    return i;
                });//发布者

//        flux.subscribe();//订阅
//        flux.subscribe((v) -> System.out.println("v == " + v));//正常消费者
//        flux.subscribe((e) -> System.out.println("e == " + e),(e) -> OutColour.out.printlnRed("错误:",e.getMessage()));//正常消费者 + 异常处理
        flux.subscribe((e) -> System.out.println("e == " + e), //正常消费者
                (e) -> OutColour.out.printlnRed("错误:",e.getMessage()),//异常处理
                () -> OutColour.out.printlnBlue("流正常结束"));//感知正常结束
        /**
         * 自定义消费者
         *  {@link MyBaseSubscriber}  继承 {@link BaseSubscriber}
         */
    }
    @Test
    public void Flux自定义订阅方法() throws IOException {
        /**
         * 自定义消费者
         *  {@link MyBaseSubscriber}  继承 {@link BaseSubscriber}
         */
        Flux<Integer> flux = Flux.range(1, 5)
                .map(i -> {
                    if (i == 3) {
                        i = i / 0;
                    }
                    return i;
                })
                ;//发布者
        flux.subscribe(new MyBaseSubscriber());//消费者‘

    }
}

class MyBaseSubscriber extends BaseSubscriber<Integer>{//自定义 订阅者（消费者）
    @Override
    protected void hookOnSubscribe(Subscription subscription) {
        System.out.println("订阅者和发布者绑定好了：" + subscription);
//        request(1);//背压
        requestUnbounded();//要所有数据，但是每个元素到达，还是会触发 hookOnNext
    }

    @Override
    protected void hookOnNext(Integer value) {
        System.out.println("元素到达：" + value);
        if (value == 3) {
            value = value / 0;
        }
        if(false){
            cancel();//取消流
            /***
             * 流能够取消，是因为有 {@link Disposable#dispose}  继承了这个接口
             */
        }

        request(1);//继续要元素
    }

    @Override
    protected void hookOnComplete() {
        //        int i = 1 / 0;//会被 hookOnError 捕捉
        System.out.println("留结束了---");
    }

    @Override
    protected void hookOnError(Throwable throwable) {//感知消费者的数据
        System.out.println("异常出错：" + throwable);
    }

    @Override
    protected void hookOnCancel() {
        System.out.println("流取消了");
    }

    @Override
    protected void hookFinally(SignalType type) {
        //相当于 try  catch 的  finally
        try {

        }catch (Exception e){

        }finally {

        }
        System.out.println("无论如何个  我都执行  ----");
    }

}